#include "execution/executors/sort_executor.h"

namespace bustub {

SortExecutor::SortExecutor(ExecutorContext *exec_ctx, const SortPlanNode *plan,
                           std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child_executor)) {}

void SortExecutor::Init() {
  child_executor_->Init();
  Tuple tuple;
  RID rid;
  while (child_executor_->Next(&tuple, &rid)) {
    tuple_vec_.push_back(tuple);
  }
  std::sort(tuple_vec_.begin(), tuple_vec_.end(), [&](Tuple &t1, Tuple &t2) {
    for (const auto &order_by : plan_->order_bys_) {
      const Value l = order_by.second->Evaluate(&t1, plan_->OutputSchema());
      const Value r = order_by.second->Evaluate(&t2, plan_->OutputSchema());
      auto equ = l.CompareEquals(r);
      if (equ == CmpBool::CmpTrue) {
        continue;
      }
      auto res = l.CompareGreaterThan(r);
      if (res == CmpBool::CmpTrue) {
        if (order_by.first == OrderByType::DESC) {
          return true;
        }
        if (order_by.first == OrderByType::ASC || order_by.first == OrderByType::DEFAULT) {
          return false;
        }
      }
      if (res == CmpBool::CmpFalse) {
        if (order_by.first == OrderByType::DESC) {
          return false;
        }
        if (order_by.first == OrderByType::ASC || order_by.first == OrderByType::DEFAULT) {
          return true;
        }
      }
    }
    return false;
  });
  tuple_iterator_ = tuple_vec_.begin();
}

auto SortExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  while (tuple_iterator_ != tuple_vec_.end()) {
    *tuple = *tuple_iterator_;
    *rid = tuple->GetRid();
    ++tuple_iterator_;
    return true;
  }
  return false;
}

}  // namespace bustub
